package com.jacky.rxjava.v315;


import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.subjects.BehaviorSubject;

import java.util.concurrent.TimeUnit;

/**
 * @program: 智灵时代广州研发中心
 * @description:
 * @author: 吞星(yangguojun)
 * @create: 2021-12-28 10:23
 **/
public class RXJaveDemo3 {


    public static void main(String[] args) {
        BehaviorSubject<String> observable = BehaviorSubject.create();
        //MILLISECONDS表示毫秒
        Observable<Observable<String>> windowObservable = observable.window(5000, TimeUnit.MILLISECONDS);
        windowObservable.subscribe(stringObservable -> {
            long start = System.currentTimeMillis();
            System.out.println("开始统计: " + start);
            final int[] count = {0};
            stringObservable.subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                }

                @Override
                public void onNext(@NonNull String s) {
                    count[0]++;
                }

                @Override
                public void onError(@NonNull Throwable e) {
                }

                @Override
                public void onComplete() {
                    System.out.println("统计完成，一共耗费: " + (System.currentTimeMillis() - start));
                    System.out.println("一共收到" + count[0] + "次");
                    System.out.println();
                }
            });
        });
        for (int i = 0; i < 1000; i++) {
            //200ms生产一个数据，
            observable.onNext("songyao");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}
